#include							"Processor.h"
#include							"BoostAsyncTcpSocket.h"
#include							"BoostRotatingBuffer.h"
#include							"Operator.h"
#include							"Protocol.h"

enum Protocol : byte;

BoostAsyncTcpSocket::BoostAsyncTcpSocket(Processor &proc, boost::asio::io_service &io_service, unsigned short id) : _proc(proc), _service(io_service), _socket(_service), _current_msg(0), _id(id), _heartbeat_timer(io_service)
{
	_buffer = new BoostRotatingBuffer;
	_buffer->init();
}

BoostAsyncTcpSocket::~BoostAsyncTcpSocket()
{
	if (_buffer)
		delete _buffer;
}

void								BoostAsyncTcpSocket::init()
{
	startRead();
	start_heartbeat();
}

void								BoostAsyncTcpSocket::sendOrder(IOrder* order)
{
	unsigned short					size = order->size();

	this->write(order->send(), size + sizeof(unsigned short));
}

size_t								BoostAsyncTcpSocket::write(byte *data, size_t size)
{
	 boost::asio::async_write(_socket, boost::asio::buffer(data, size),
          boost::bind(&BoostAsyncTcpSocket::handle_write, this,
            boost::asio::placeholders::error));
	return (0);
}

void								BoostAsyncTcpSocket::handle_write(const boost::system::error_code& ec)
{

}

void								BoostAsyncTcpSocket::close()
{
	_service.post(boost::bind(&BoostAsyncTcpSocket::DoClose, this));
}

void								BoostAsyncTcpSocket::OnConnect(const boost::system::error_code& ErrorCode, boost::asio::ip::tcp::resolver::iterator EndPointIter)
{
	std::cout << "connect hook" << std::endl;
	if (!_socket.is_open())
	{
		std::cout << "Connect timed out\n";
		EndPointIter++;
	}
	else if (ErrorCode)
	{
		std::cout << "Connect error: " << ErrorCode.message() << "\n";
		_socket.close();
	}
	else
	{
		std::cout << "Connected to " << EndPointIter->endpoint() << "\n";
		startRead();
	}
}

void								BoostAsyncTcpSocket::startRead()
{
	_socket.async_receive(boost::asio::buffer(_tmp_buf, SOCKET_READ_LENGTH), boost::bind(&BoostAsyncTcpSocket::OnReceive, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
}

void								BoostAsyncTcpSocket::OnReceive(const boost::system::error_code& error, size_t bytes_transferred)
{
	if (!error)
	{
		for (size_t cnt = 0; cnt < bytes_transferred; ++cnt)
			_buffer->write(_tmp_buf[cnt]);
		if (!_current_msg && _buffer->size() >= 2)
			getOrderSize();
		while (_buffer->size() > 0 && _buffer->size() >= _current_msg)
		{
			int size = _buffer->size();
			std::vector<byte>		*harvest = new std::vector<byte>;
			for (unsigned short cnt = 0; cnt < _current_msg; ++cnt)
				harvest->push_back(_buffer->read());
			_proc(harvest, _id);
			_current_msg = 0;
			if (_buffer->size() >= 2)
				getOrderSize();
		}
		startRead();
	}
	else
		DoClose();
}

void								BoostAsyncTcpSocket::getOrderSize()
{
	byte size[2];

	size[0] = _buffer->read();
	size[1] = _buffer->read();
	_current_msg = *((unsigned short*) size);
}

void								BoostAsyncTcpSocket::OnSend(const boost::system::error_code& ErrorCode)
{

}

void								BoostAsyncTcpSocket::DoClose()
{
	_socket.close();
}

void								BoostAsyncTcpSocket::connectToHost(const std::string & ip, const std::string & port)
{
	boost::asio::ip::tcp::resolver resolver(_service);
	boost::asio::ip::tcp::resolver::query query(boost::asio::ip::tcp::v4(), ip, port);
	boost::asio::ip::tcp::resolver::iterator iterator = resolver.resolve(query);
	std::cout << "Connecting to " << iterator->endpoint() << std::endl;
	_socket.async_connect(iterator->endpoint(), boost::bind(&BoostAsyncTcpSocket::OnConnect, this, _1, iterator));
}

boost::asio::ip::tcp::socket		&BoostAsyncTcpSocket::getSocket()
{
	return _socket;
}

void								BoostAsyncTcpSocket::start_heartbeat()
{
	Order							order;
	Protocol						protocol = SERVER;
	Operator						op = PING;
	boost::chrono::duration<double> sec = boost::chrono::nanoseconds(_proc.time.elapsed().user);
	unsigned int					ts = (unsigned int)(sec.count() * 1000);

	std::cout << "sent heartbeat with timestamp " << ts << std::endl;
	order.init();
	order.pushData(&protocol, sizeof(protocol));
	order.pushData(&op, sizeof(op));
	order.pushData(&ts, sizeof(ts));
	boost::asio::async_write(_socket, boost::asio::buffer(order.send(), order.size() + sizeof(unsigned short)),
    boost::bind(&BoostAsyncTcpSocket::handle_heartbeat, this,
    boost::asio::placeholders::error));
}

 void								BoostAsyncTcpSocket::handle_heartbeat(const boost::system::error_code& ec)
{
    if (!ec)
    {
      _heartbeat_timer.expires_from_now(boost::posix_time::seconds(4));
      _heartbeat_timer.async_wait(boost::bind(&BoostAsyncTcpSocket::start_heartbeat, this));
    }
    else
    {
      std::cout << "Error on heartbeat: " << ec.message() << "\n";
	  DoClose();
    }
}

std::string							BoostAsyncTcpSocket::getIp()
{
	return _socket.remote_endpoint().address().to_string();
}
